package com.twang.stock.mq.rocketMq;


import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.twang.stock.Entity.ShiShiData;
import com.twang.stock.common.http.HtmlTools;
import com.twang.stock.dto.EbayRequest;
import com.twang.stock.dto.EmailMessage;
import com.twang.stock.dto.StockMinutesMessage;
import com.twang.stock.ebay.EbayService;
import com.twang.stock.entity.CompanyStockMinutes;
import com.twang.stock.service.*;
import com.twang.stock.service.jinyan.StockMinutesService;
import com.twang.stock.service.jinyan.UpdateStockCoreService;
import com.twang.stock.strategy.OneMinutesTradeSum;
import com.twang.stock.strategy.Stock60MinutesUpdate;
import com.twang.stock.threadPool.AsycTaskExecutorConfig;
import com.twang.stock.utils.date.HolidayCopy;
import com.twang.stock.zq.ZhaiQuanServcie;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.net.InetAddress;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 设置消息监听
 * 监听组：监听topic：监听tag(默认监听topic下所有)
 * 监听消费模式:默认负载均衡：CLUSTERING（每一个消息只发给一个消费者）、广播模式：BROADCASTING（发送给所有消费者）
 *
 */
@Service
@RocketMQMessageListener(topic = "${rocketmq.consumer.topic}", consumerGroup = "${rocketmq.consumer.group}",selectorExpression = "${rocketmq.consumer.tags}")
@Slf4j
public class RocketMQConsumer  implements RocketMQListener<String> {


    @Autowired
    private WeiPanJianQianService weiPanJianQianService;
    @Autowired
    private StockMinutesService stockMinutesService;

    @Autowired
    private AsycTaskExecutorConfig asycTaskExecutorConfig;

    @Autowired
    private MinutesStockFixService minutesStockFixService;
    @Resource
    private MinutesChengJIaoeService minutesChengJIaoeService;

    @Autowired
    private ZuoTiService zuoTiService;
    @Autowired
    private NotifyStockBuyService notifyStockBuyService;

    @Autowired
    private WanZhuBeiChiChangService wanZhuBeiChiChangService;

    @Autowired
    private ZhangTingjuJiaoService zhangTingjuJiaoService;

    @Autowired
    private OneMinutesTradeSum oneMinutesTradeSum;
    @Autowired
    private Stock60MinutesUpdate stock60MinutesUpdate;
    @Autowired
    private KaiPanLaService kaiPanLaService;
    @Autowired
    private EmailWranConsumer emailWranConsumer;

    @Autowired
    private ZhaiQuanServcie zhaiQuanServcie;

    @Autowired
    private EbayService ebayService;


    @Override
    public void onMessage(String message) {
        log.info(">>>>>>>>>>> 消费到消息： value={}",message);
        try {
            receiveMessageNext(message);
        } catch (Exception e) {
            log.error(">>>>>>>>>>> 消费到消息： value={}",message);
        }
    }

    private void receiveMessageNext(String message) throws Exception {
        log.info("==========> 接收到消息:direct.stock.queue-{}  接收到消息:{} ", InetAddress.getLocalHost().getHostAddress(), message);
        if (StringUtils.isNotEmpty(message)) {
            StockMinutesMessage stockMinutesMessage = JSONObject.parseObject(message, StockMinutesMessage.class);
            if (StringUtils.isEmpty(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage  未知数据类型,message={} ", JSONObject.toJSONString(stockMinutesMessage));
            }
            if (stockMinutesMessage.getType().length() == 1) {
                log.info("==========>StockMiuntesReceiver  receiveMessage  邮件预警,message={}  type={}", JSONObject.toJSONString(stockMinutesMessage), stockMinutesMessage.getType());
                //发送mail
                emailWranConsumer.receiveMessage(message);
            }
            if (StockMinutesMessage.TYPE_MINUTES.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage  分时处理 start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                excute(stockMinutesMessage);
                //
            }
            if (StockMinutesMessage.TYPE_MINUTES_FIX.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage  分时修复 start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                minutesStockFixService.initMinutesData(stockMinutesMessage);
                wanZhuBeiChiChangService.根据日期更新玩主杯热榜(HolidayCopy.getToday());
            }
            if (StockMinutesMessage.TYPE_MINUTES_CHENG_JIAO_E.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage  分时成交额 start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                    minutesChengJIaoeService.excute();
                });
            }
            if (StockMinutesMessage.TYPE_MINUTES_ZUO_T.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage  做T预警触发 start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                    zuoTiService.做T预警();
                });
            }
            //主页更新-1分钟一次
            if (StockMinutesMessage.TYPE_MINUTES_INDEX_INFO.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage  刷新主页信息 start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                    notifyStockBuyService.execute();
                });
                String minutes = stockMinutesMessage.getMinutes();
                if (!"0930".equals(minutes) || !"0931".equals(minutes) || !"0932".equals(minutes) || !"0933".equals(minutes)
                        || !"0934".equals(minutes) || !"0935".equals(minutes)) {
                    asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                        try {
                            log.info(">>>>>>>>> 转债 执行监控预警 date={} minutes={}", stockMinutesMessage.getDate(), stockMinutesMessage.getMinutes());
                            zhaiQuanServcie.getzhuanzhaiList(stockMinutesMessage.getDate(), stockMinutesMessage.getMinutes());
                        } catch (InterruptedException e) {
                            log.error(">>>>>>>>> 转债 执行监控预警异常 e", e);
                        }
                    });
                }

            }
            //涨停聚焦
            if (StockMinutesMessage.TYPE_ZHANG_TING_JU_JIAO.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage 涨停聚焦  start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                    zhangTingjuJiaoService.涨停聚焦(stockMinutesMessage.getDate());
                });
            }

            //60日成交量预警
            if (StockMinutesMessage.TYPE_ONE_MINUTES_CJE_WARN.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage 分时超过1分钟预警  start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                String[] codes = stockMinutesMessage.getCode().split(",");
                List<ShiShiData> shiShiDataList = HtmlTools.getGpdatas(Lists.newArrayList(codes));
                for (ShiShiData shiShiData : shiShiDataList) {
                    asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                        oneMinutesTradeSum.max60DayTradeVol(shiShiData);
                    });
                }
                log.info("==========>StockMiuntesReceiver 分时超过1分钟预警 and 60天最大成交量  warn completed  num={} ", codes.length);

            }

            //60天最大成交量update
            if (StockMinutesMessage.TYPE_60_DAY_MAX_VOL_UPDATE.equals(stockMinutesMessage.getType())) {
                log.info("==========>StockMiuntesReceiver  receiveMessage 60天最大成交量update  start IP：{}  code:{} minutes:{} ", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
                String[] codes = stockMinutesMessage.getCode().split(",");
                CountDownLatch countDownLatch = new CountDownLatch(codes.length);
                for (String code : codes) {
                    asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                        try {
                            stock60MinutesUpdate.updateMaxVol(code);
                        } finally {
                            countDownLatch.countDown();
                            log.info("==========>StockMiuntesReceiver 60天最大成交量 update left={}", countDownLatch.getCount());
                        }
                    });
                }
                countDownLatch.await();
            }
            //涨停板连扳数据
            if (StockMinutesMessage.TYPE_ZHANG_TING_BAN_LIAN_BAN.equals(stockMinutesMessage.getType())) {
                kaiPanLaService.生成今日涨停信息();
            }

            //开盘啦 实时龙虎榜数据
            if (StockMinutesMessage.TYPE_ZI_JING_JIN_E.equals(stockMinutesMessage.getType())) {
                asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                    try {
                        log.info(">>>>>>>>>> 开盘啦-实时龙虎榜数据 info={}", JSONObject.toJSONString(stockMinutesMessage));
                        kaiPanLaService.实时龙虎榜数据(stockMinutesMessage.getMinutes());
                    } catch (Exception e) {
                        log.error(">>>>>>>>>> 开盘啦-实时龙虎榜数据 error={}", e);
                        throw new RuntimeException(e);
                    }
                });
            }
            //开盘啦bankuai
            if (StockMinutesMessage.TYPE_BAN_KUAI_JIN_E.equals(stockMinutesMessage.getType())) {
                asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                    try {
                        log.info(">>>>>>>>>> 开盘啦-开盘啦bankuai info={}", JSONObject.toJSONString(stockMinutesMessage));
                        kaiPanLaService.createTodayBanKuaiInfobyMinutes(stockMinutesMessage.getMinutes());
                    } catch (Exception e) {
                        log.error(">>>>>>>>>> 开盘啦-开盘啦bankuai error={}", e);
                        throw new RuntimeException(e);
                    }
                });
            }

            //尾盘买，开盘卖
            if (StockMinutesMessage.TYPE_WARN_WEI_PAN_JIAN_QIAN.equals(stockMinutesMessage.getType())) {
                asycTaskExecutorConfig.taskQueueExecutor().execute(() -> {
                    try {
                        log.info(">>>>>>>>>> 尾盘 执行选股 info={}", JSONObject.toJSONString(stockMinutesMessage));
                        weiPanJianQianService.execute("");
                    } catch (Exception e) {
                        log.error(">>>>>>>>> 尾盘 执行选股 error={}", e);
                        throw new RuntimeException(e);
                    }
                });
            }

            //入库DB削峰
            if (StockMinutesMessage.TYPE_INSERT_MINUTES_DB.equals(stockMinutesMessage.getType())) {
                log.info(">>>>>>>>>> 入库DB削峰 name={} ", stockMinutesMessage.getCode());
                minutesStockFixService.insertMqBatch((List<CompanyStockMinutes>) stockMinutesMessage.getData());
            }

            //入库DB削峰
            if (StockMinutesMessage.TYPE_ITEM_HOME_PAGE.equals(stockMinutesMessage.getType())) {
                log.info(">>>>>>>>>> 生成搜收页面的商品 name={} ", stockMinutesMessage.getCode());
                //EbayRequest ebayRequest = (EbayRequest) stockMinutesMessage.getData();
                EbayRequest ebayRequest = JSONObject.parseObject(JSONObject.toJSONString(stockMinutesMessage.getData()), EbayRequest.class);
                try{
                    ebayService.itemUserHomePage(ebayRequest.getGoodsUrl(), ebayRequest.getName());
                }catch (Exception e){
                   // ebayService.ebayItemSendToMq(ebayRequest.getGoodsUrl(), ebayRequest.getName());
                }
            }
        }
    }

    public void excute(StockMinutesMessage stockMinutesMessage) {
        try {
            stockMinutesService.createStockMinutesDataByCode(stockMinutesMessage);
            log.info("==========> direct.stock.queue IP：{}  {} {} 分时处理结束", InetAddress.getLocalHost().getHostAddress(), stockMinutesMessage.getCode(), stockMinutesMessage.getMinutes());
        } catch (Exception e) {
            log.error("{}", e);
        }
    }


}
